from kafka import KafkaConsumer
import json
import re
import struct

# Kafka配置
KAFKA_SERVER = '192.168.162.118:9092'
TOPIC_NAME = 'LAB-HTTP'

def advanced_deserialize(m):
    """
    高级反序列化函数，尝试多种解析方法
    """
    if not m:
        return ""

    # 尝试直接UTF-8解码
    try:
        decoded = m.decode('utf-8')
        return decoded
    except UnicodeDecodeError:
        pass

    # 尝试查找结构化数据
    try:
        # 尝试作为字节数据解析
        decoded = m.decode('latin-1')
        
        # 检查是否包含可识别的模式
        if 'Header:' in decoded and 'HTTP/' in decoded:
            return decoded
            
        # 检查是否是纯HTTP数据
        if decoded.startswith(('GET ', 'POST ', 'PUT ', 'DELETE ', 'HTTP/')):
            return decoded
            
        # 如果没有识别出模式，返回原始十六进制
        return f"[Raw bytes] {m.hex()}"
    except Exception as e:
        return f"[Deserialization Error: {str(e)}]"

def safe_key_deserialize(key):
    """安全的key反序列化函数"""
    if not key:
        return ""
    try:
        return key.decode('utf-8')
    except UnicodeDecodeError:
        return f"[Binary Key] {key.hex()}"
    except Exception as e:
        return f"[Key Deserialization Error: {str(e)}]"

# 创建Kafka消费者
print("正在初始化Kafka消费者...")
try:
    consumer = KafkaConsumer(
        TOPIC_NAME,
        bootstrap_servers=[KAFKA_SERVER],
        value_deserializer=advanced_deserialize,
        key_deserializer=safe_key_deserialize,
        auto_offset_reset='latest',
        enable_auto_commit=True,
        consumer_timeout_ms=30000,  # 30秒超时
    )
    print(f"Kafka消费者初始化成功")
except Exception as e:
    print(f"Kafka消费者初始化失败: {e}")
    exit(1)

print(f"开始监听Kafka topic: {TOPIC_NAME}，服务器: {KAFKA_SERVER}")
print("注意: Kafka Tool可能使用了特定的插件(dpi-kafka-plugin)来解析消息格式")
print("我们正在尝试以最接近的方式显示原始数据...\n")

try:
    message_count = 0
    max_messages = 3  # 读取3条消息以获取更多信息

    for message in consumer:
        message_count += 1
        print(f"\n{'='*80}")
        print(f"第 {message_count} 条消息")
        print(f"{'='*80}")
        print(f"分区: {message.partition}")
        print(f"偏移量: {message.offset}")
        
        # 显示key
        if message.key:
            print(f"Key: {message.key}")
        
        # 显示value
        if message.value:
            print("Value:")
            print("-" * 40)
            # 直接打印，不进行额外处理
            print(repr(message.value)[1:-1])  # 去掉字符串的引号
            print("-" * 40)
            
            # 如果看起来像包含Header的数据，尝试解析
            if isinstance(message.value, str) and 'Header:' in message.value:
                print("\n[解析信息]")
                lines = message.value.split('\n')
                header_line = None
                for line in lines:
                    if line.startswith('Header:'):
                        header_line = line
                        break
                
                if header_line:
                    print(f"检测到Header行: {header_line}")
        else:
            print("Value: (空)")

        # 达到指定条数后停止
        if message_count >= max_messages:
            print(f"\n已读取 {max_messages} 条消息，停止监听")
            break

except Exception as e:
    print(f"Kafka消费者出错: {e}")
    import traceback
    traceback.print_exc()
finally:
    consumer.close()
    print(f"\nKafka消费者已关闭")